package com.hui.qiniucloud.service.impl;

import com.hui.qiniucloud.config.TtsProperties;
import com.hui.qiniucloud.service.ITtsV3Service;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;

import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.io.*;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.JsonObject;
import com.google.gson.JsonArray;

@Slf4j
@Service
public class TtsV3Service implements ITtsV3Service {

    // V3双向流式TTS API端点
    private static final String WSS_ENDPOINT = "wss://openspeech.bytedance.com/api/v3/tts/bidirection";

    // 协议常量 - 基于V3 API规范
    private static final byte PROTOCOL_VERSION = 0b0001;
    private static final byte DEFAULT_HEADER_SIZE = 0b0001;
    
    // Message Type
    private static final byte FULL_CLIENT_REQUEST = 0b0001;
    private static final byte FULL_SERVER_RESPONSE = 0b1001;
    private static final byte SERVER_ERROR_RESPONSE = 0b1111;
    
    // Message Type Specific Flags
    private static final byte EVENT_FLAG = 0b0100;
    
    // Serialization Type
    private static final byte NO_SERIALIZATION = 0b0000;
    private static final byte JSON = 0b0001;
    
    // Compression Type
    private static final byte NO_COMPRESSION = 0b0000;
    private static final byte GZIP = 0b0001;

    // Event codes - TTS V3 Events
    private static final int START_CONNECTION = 1;
    private static final int FINISH_CONNECTION = 2;
    private static final int CONNECTION_STARTED = 50;
    private static final int CONNECTION_FAILED = 51;
    private static final int CONNECTION_FINISHED = 52;
    private static final int START_SESSION = 100;
    private static final int CANCEL_SESSION = 101;
    private static final int FINISH_SESSION = 102;
    private static final int SESSION_STARTED = 150;
    private static final int SESSION_CANCELED = 151;
    private static final int SESSION_FINISHED = 152;
    private static final int SESSION_FAILED = 153;
    private static final int USAGE_RESPONSE = 154;
    private static final int TASK_REQUEST = 200;
    private static final int TTS_SENTENCE_START = 350;
    private static final int TTS_SENTENCE_END = 351;
    private static final int TTS_RESPONSE = 352;

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Map<String, WebSocketClient> activeConnections = new ConcurrentHashMap<>();

    @Autowired
    private TtsProperties ttsProperties;

    @PostConstruct
    public void validateConfig() {
        if (!StringUtils.hasText(ttsProperties.getAppId())) {
            log.error("TTS配置错误: appId 为空，请检查 volcengine.tts.app-id 配置");
        }
        if (!StringUtils.hasText(ttsProperties.getToken())) {
            log.error("TTS配置错误: token 为空，请检查 volcengine.tts.token 配置");
        }

        log.info("V3 TTS配置加载成功: appId={}, endpoint={}", ttsProperties.getAppId(), ttsProperties.getEndpoint());
    }

    /**
     * WebSocket流式文字转语音 - V3双向流式API
     * @param request TTS请求参数
     * @return 音频数据流
     */
    public Flux<ITtsV3Service.TtsResponse> synthesizeSpeech(ITtsV3Service.TtsRequest request) {
        // 验证配置
        if (!StringUtils.hasText(ttsProperties.getAppId()) ||
            !StringUtils.hasText(ttsProperties.getToken())) {
            log.error("TTS配置不完整，无法进行语音合成");
            return Flux.empty();
        }

        // 验证文本
        if (!StringUtils.hasText(request.getText())) {
            log.error("文本内容为空");
            return Flux.empty();
        }

        log.info("开始V3双向流式TTS合成，文本: '{}'", request.getText());

        String connectId = UUID.randomUUID().toString();
        
        return Flux.create(sink -> {
            try {
                WebSocketClient client = createV3TtsWebSocketClient(connectId, request, sink);
                activeConnections.put(connectId, client);
                client.connect();
                
            } catch (Exception e) {
                log.error("创建TTS WebSocket连接失败", e);
                sink.error(e);
            }
        });
    }

    private WebSocketClient createV3TtsWebSocketClient(String connectId, ITtsV3Service.TtsRequest request, 
                                                       reactor.core.publisher.FluxSink<ITtsV3Service.TtsResponse> sink) {
        try {
            URI serverUri = new URI(ttsProperties.getEndpoint());
            AtomicInteger currentStage = new AtomicInteger(0); // 0: 建连, 1: 会话, 2: 任务
            
            WebSocketClient client = new WebSocketClient(serverUri) {
                @Override
                public void onOpen(ServerHandshake handshake) {
                    log.info("V3 TTS WebSocket连接已建立: {}, 状态码: {}", connectId, handshake.getHttpStatus());
                    String logId = handshake.getFieldValue("X-Tt-Logid");
                    if (logId != null) {
                        log.info("X-Tt-Logid: {}", logId);
                    }

                    try {
                        // 1. 发送 StartConnection
                        sendStartConnection(this);
                        currentStage.set(0);
                        
                    } catch (Exception e) {
                        log.error("发送StartConnection失败", e);
                        sink.error(e);
                        if (isOpen()) close();
                    }
                }

                @Override
                public void onMessage(String message) {
                    log.debug("收到文本消息: {}", message);
                }

                @Override
                public void onMessage(ByteBuffer bytes) {
                    try {
                        log.debug("收到V3二进制消息，长度: {} bytes", bytes.remaining());
                        TtsV3Response response = parseV3TtsBinaryResponse(bytes.array());
                        
                        if (response != null) {
                            log.info("V3 TTS响应: event={}, hasAudio={}", response.event, response.audioData != null);
                            
                            // 处理不同事件
                            switch (response.event) {
                                case CONNECTION_STARTED:
                                    log.info("连接建立成功，开始会话");
                                    sendStartSession(this, request);
                                    currentStage.set(1);
                                    break;
                                    
                                case SESSION_STARTED:
                                    log.info("会话开始成功，发送任务请求");
                                    sendTaskRequest(this, request);
                                    currentStage.set(2);
                                    break;
                                    
                                case TTS_SENTENCE_START:
                                    log.info("句子开始: {}", response.text);
                                    break;
                                    
                                case TTS_RESPONSE:
                                    if (response.audioData != null && response.audioData.length > 0) {
                                        log.info("收到音频数据: {} bytes", response.audioData.length);
                                        sink.next(new ITtsV3Service.TtsResponse(response.audioData, response.text, response.event, false));
                                    }
                                    break;
                                    
                                case TTS_SENTENCE_END:
                                    log.info("句子结束: {}", response.text);
                                    break;
                                    
                                case SESSION_FINISHED:
                                    log.info("会话结束");
                                    sendFinishConnection(this);
                                    break;
                                    
                                case CONNECTION_FINISHED:
                                    log.info("连接结束");
                                    sink.next(new ITtsV3Service.TtsResponse(null, null, response.event, true));
                                    sink.complete();
                                    if (isOpen()) close();
                                    break;
                                    
                                case CONNECTION_FAILED:
                                case SESSION_FAILED:
                                    log.error("TTS服务失败: event={}, message={}", response.event, response.text);
                                    sink.error(new RuntimeException("TTS服务失败: " + response.text));
                                    if (isOpen()) close();
                                    break;
                            }
                        }
                        
                    } catch (Exception e) {
                        log.error("解析V3 TTS响应失败", e);
                        sink.error(e);
                    }
                }

                @Override
                public void onClose(int code, String reason, boolean remote) {
                    log.warn("V3 TTS WebSocket连接已关闭: {} - {} (remote: {})", code, reason, remote);
                    if (code == 1006) {
                        log.error("连接异常关闭，可能是认证失败或协议错误");
                    }
                    sink.complete();
                }

                @Override
                public void onError(Exception ex) {
                    log.error("V3 TTS WebSocket连接错误: {}", ex.getMessage(), ex);
                    sink.error(ex);
                }
            };

            // V3双向流式TTS API认证
            client.addHeader("X-Api-App-Key", ttsProperties.getAppId());
            client.addHeader("X-Api-Access-Key", ttsProperties.getToken());
            client.addHeader("X-Api-Resource-Id", ttsProperties.getResourceId());
            client.addHeader("X-Api-Connect-Id", connectId);
            client.addHeader("User-Agent", "qiniucloud-tts-v3/2.0");

            log.debug("V3 TTS WebSocket 认证头信息:");
            log.debug("  X-Api-App-Key: {}", ttsProperties.getAppId());
            log.debug("  X-Api-Access-Key: {}", ttsProperties.getToken() != null ? "***" + ttsProperties.getToken().substring(Math.max(0, ttsProperties.getToken().length() - 4)) : "null");
            log.debug("  X-Api-Resource-Id: {}", ttsProperties.getResourceId());
            log.debug("  X-Api-Connect-Id: {}", connectId);

            // 添加详细的参数检查日志
            if (ttsProperties.getAppId() == null || ttsProperties.getAppId().equals("your-app-id")) {
                log.error("❌ VOLCENGINE_TTS_APP_ID 环境变量未设置或使用默认值");
            }
            if (ttsProperties.getToken() == null || ttsProperties.getToken().equals("your-token")) {
                log.error("❌ VOLCENGINE_TTS_TOKEN 环境变量未设置或使用默认值");
            }
            if (ttsProperties.getResourceId() == null) {
                log.error("❌ TTS Resource ID 未设置");
            }

            return client;

        } catch (Exception e) {
            throw new RuntimeException("创建V3 TTS WebSocket客户端失败", e);
        }
    }

    /**
     * TTS V3响应解析结果类
     */
    static class TtsV3Response {
        public int event;
        public String text;
        public byte[] audioData;
        
        @Override
        public String toString() {
            return String.format("TtsV3Response{event=%d, text='%s', audioSize=%d}", 
                event, text, audioData != null ? audioData.length : 0);
        }
    }

    /**
     * 解析V3 TTS二进制响应
     */
    private TtsV3Response parseV3TtsBinaryResponse(byte[] responseData) {
        if (responseData == null || responseData.length == 0) {
            return null;
        }

        TtsV3Response result = new TtsV3Response();

        try {
            // 解析头部
            int protocolVersion = (responseData[0] >> 4) & 0x0f;
            int headerSize = responseData[0] & 0x0f;
            int messageType = (responseData[1] >> 4) & 0x0f;
            int messageTypeSpecificFlags = responseData[1] & 0x0f;
            int serializationMethod = (responseData[2] >> 4) & 0x0f;
            int messageCompression = responseData[2] & 0x0f;

            log.debug("V3 TTS响应头部 - 协议版本: {}, 头大小: {}, 消息类型: {}, 标志: {}, 序列化: {}, 压缩: {}",
                     protocolVersion, headerSize, messageType, messageTypeSpecificFlags, serializationMethod, messageCompression);

            // 解析payload
            byte[] payload = java.util.Arrays.copyOfRange(responseData, headerSize * 4, responseData.length);
            
            // 检查是否包含事件号
            if ((messageTypeSpecificFlags & 0x04) != 0) {
                result.event = bytesToInt(java.util.Arrays.copyOfRange(payload, 0, 4));
                payload = java.util.Arrays.copyOfRange(payload, 4, payload.length);
            }

            // 根据消息类型处理
            if (messageType == FULL_SERVER_RESPONSE) {
                // 完整服务器响应 - 跳过payload size
                if (payload.length >= 4) {
                    int payloadSize = bytesToInt(java.util.Arrays.copyOfRange(payload, 0, 4));
                    payload = java.util.Arrays.copyOfRange(payload, 4, payload.length);
                }
            }

            if (payload.length == 0) {
                return result;
            }

            // 解压缩
            if (messageCompression == GZIP) {
                payload = gzipDecompress(payload);
            }

            // 根据序列化方法解析
            if (serializationMethod == JSON && payload != null) {
                String jsonStr = new String(payload, StandardCharsets.UTF_8);
                result.text = jsonStr;
                log.debug("收到JSON响应: {}", jsonStr);
            } else if (serializationMethod == NO_SERIALIZATION && payload != null) {
                result.audioData = payload;
                log.debug("收到原始音频数据: {} bytes", payload.length);
            }

        } catch (Exception e) {
            log.error("解析V3 TTS二进制响应失败", e);
        }

        return result;
    }

    /**
     * 发送StartConnection
     */
    private void sendStartConnection(WebSocketClient client) throws Exception {
        JsonObject payload = new JsonObject();
        payload.addProperty("event", START_CONNECTION);
        
        sendV3Message(client, payload, START_CONNECTION);
        log.info("发送StartConnection成功");
    }

    /**
     * 发送StartSession
     */
    private void sendStartSession(WebSocketClient client, ITtsV3Service.TtsRequest request) throws Exception {
        JsonObject user = new JsonObject();
        user.addProperty("uid", "qiniucloud_uid");

        JsonObject audioParams = new JsonObject();
        audioParams.addProperty("format", request.getFormat());
        audioParams.addProperty("sample_rate", request.getSampleRate());
        audioParams.addProperty("bit_rate", request.getBitRate());
        audioParams.addProperty("loudness_rate", request.getLoudnessRate());
        audioParams.addProperty("speed", request.getSpeed());

        JsonObject reqParams = new JsonObject();
        reqParams.addProperty("speaker", request.getSpeaker() != null ? request.getSpeaker() : ttsProperties.getDefaultSpeaker());
        reqParams.add("audio_params", audioParams);

        // 如果有混音配置
        if (request.getMixSpeaker() != null) {
            reqParams.addProperty("speaker", "custom_mix_bigtts");
            
            JsonObject mixSpeaker = new JsonObject();
            JsonArray speakers = new JsonArray();
            
            for (ITtsV3Service.SpeakerMix speakerMix : request.getMixSpeaker().getSpeakers()) {
                JsonObject speaker = new JsonObject();
                speaker.addProperty("source_speaker", speakerMix.getSourceSpeaker());
                speaker.addProperty("mix_factor", speakerMix.getMixFactor());
                speakers.add(speaker);
            }
            mixSpeaker.add("speakers", speakers);
            reqParams.add("mix_speaker", mixSpeaker);
        }

        JsonObject payload = new JsonObject();
        payload.add("user", user);
        payload.addProperty("event", START_SESSION);
        payload.add("req_params", reqParams);

        sendV3Message(client, payload, START_SESSION);
        log.info("发送StartSession成功");
    }

    /**
     * 发送TaskRequest
     */
    private void sendTaskRequest(WebSocketClient client, ITtsV3Service.TtsRequest request) throws Exception {
        JsonObject user = new JsonObject();
        user.addProperty("uid", "qiniucloud_uid");

        JsonObject reqParams = new JsonObject();
        reqParams.addProperty("text", request.getText());

        JsonObject payload = new JsonObject();
        payload.add("user", user);
        payload.addProperty("event", TASK_REQUEST);
        payload.add("req_params", reqParams);

        sendV3Message(client, payload, TASK_REQUEST);
        log.info("发送TaskRequest成功，文本: '{}'", request.getText());
        
        // 发送完文本后立即发送FinishSession
        Thread.sleep(100); // 短暂延迟确保消息发送
        sendFinishSession(client);
    }

    /**
     * 发送FinishSession
     */
    private void sendFinishSession(WebSocketClient client) throws Exception {
        JsonObject payload = new JsonObject();
        payload.addProperty("event", FINISH_SESSION);
        
        sendV3Message(client, payload, FINISH_SESSION);
        log.info("发送FinishSession成功");
    }

    /**
     * 发送FinishConnection
     */
    private void sendFinishConnection(WebSocketClient client) throws Exception {
        JsonObject payload = new JsonObject();
        payload.addProperty("event", FINISH_CONNECTION);
        
        sendV3Message(client, payload, FINISH_CONNECTION);
        log.info("发送FinishConnection成功");
    }

    /**
     * 发送V3消息
     */
    private void sendV3Message(WebSocketClient client, JsonObject payload, int event) throws Exception {
        String payloadStr = payload.toString();
        byte[] payloadBytes = gzipCompress(payloadStr.getBytes(StandardCharsets.UTF_8));
        byte[] header = getV3Header(FULL_CLIENT_REQUEST, EVENT_FLAG, JSON, GZIP);
        byte[] eventBytes = intToBytes(event);
        byte[] payloadSize = intToBytes(payloadBytes.length);

        byte[] fullMessage = new byte[header.length + eventBytes.length + payloadSize.length + payloadBytes.length];
        System.arraycopy(header, 0, fullMessage, 0, header.length);
        System.arraycopy(eventBytes, 0, fullMessage, header.length, eventBytes.length);
        System.arraycopy(payloadSize, 0, fullMessage, header.length + eventBytes.length, payloadSize.length);
        System.arraycopy(payloadBytes, 0, fullMessage, header.length + eventBytes.length + payloadSize.length,
                payloadBytes.length);
        
        client.send(fullMessage);
        log.debug("V3消息发送成功，总长度: {} bytes", fullMessage.length);
    }

    // V3协议辅助方法
    private byte[] getV3Header(byte messageType, byte messageTypeSpecificFlags,
            byte serialMethod, byte compressionType) {
        final byte[] header = new byte[4];
        header[0] = (byte) ((PROTOCOL_VERSION << 4) | DEFAULT_HEADER_SIZE);
        header[1] = (byte) ((messageType << 4) | messageTypeSpecificFlags);
        header[2] = (byte) ((serialMethod << 4) | compressionType);
        header[3] = 0; // reserved
        return header;
    }

    private byte[] intToBytes(int a) {
        return new byte[] {
                (byte) ((a >> 24) & 0xFF),
                (byte) ((a >> 16) & 0xFF),
                (byte) ((a >> 8) & 0xFF),
                (byte) (a & 0xFF)
        };
    }

    private int bytesToInt(byte[] src) {
        if (src == null || (src.length != 4)) {
            throw new IllegalArgumentException("Invalid byte array for int conversion");
        }
        return ((src[0] & 0xFF) << 24)
                | ((src[1] & 0xff) << 16)
                | ((src[2] & 0xff) << 8)
                | ((src[3] & 0xff));
    }

    private byte[] gzipCompress(byte[] src) {
        if (src == null || src.length == 0) {
            return new byte[0];
        }
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        try (GZIPOutputStream gzip = new GZIPOutputStream(out)) {
            gzip.write(src);
        } catch (IOException e) {
            log.error("GZIP压缩失败", e);
            return new byte[0];
        }
        return out.toByteArray();
    }

    private byte[] gzipDecompress(byte[] src) {
        if (src == null || src.length == 0) {
            return null;
        }
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        ByteArrayInputStream ins = new ByteArrayInputStream(src);
        try (GZIPInputStream gzip = new GZIPInputStream(ins)) {
            byte[] buffer = new byte[256];
            int len;
            while ((len = gzip.read(buffer)) > 0) {
                out.write(buffer, 0, len);
            }
        } catch (IOException e) {
            log.error("GZIP解压缩失败", e);
            return null;
        }
        return out.toByteArray();
    }
}